package com.video.config;

import com.common.domain.po.Video;
import com.common.domain.po.VideoItem;
import com.common.exception.CodeAndMsg;
import com.common.exception.JavaSmException;
import com.video.service.VideoItemService;
import com.video.service.VideoService;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.List;

/**
 * @author 路亚
 * @version 0.1
 * @className RocketmqConfig
 * @description
 * @date 2024/12/26 10:56
 * @since jdk11
 */
@Configuration
public class RocketmqConfig {

    @Value("${luya.mq.name_srv_addr}")
    private String nameSrvAddr;

    @Bean
    public DefaultMQProducer defaultMQProducer(){
        DefaultMQProducer defaultMQProducer = new DefaultMQProducer("videoBrowseRecord");
        defaultMQProducer.setNamesrvAddr(nameSrvAddr);
        try {
            defaultMQProducer.start();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
        return defaultMQProducer;
    }

    @Bean
    public DefaultMQPushConsumer defaultMQPushConsumer(VideoItemService videoItemService){
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("videoLikedRecord");
        consumer.setNamesrvAddr(nameSrvAddr);
        try {
            consumer.subscribe("videoLikedRecord", "*");
            consumer.registerMessageListener(new MessageListenerConcurrently() {
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                    try {
                        for (MessageExt messageExt : list) {
                            String labelValue = new String(messageExt.getUserProperty("myLabel"));
                            String str = new String(messageExt.getBody());
                            Long videoItemId = Long.parseLong(str);
                            VideoItem videoItem = videoItemService.getById(videoItemId);
                            if (videoItem == null) {
                                throw new JavaSmException(CodeAndMsg.VIDEO_ITEM_NOT_EXIST);
                            }
                            if ("add".equals(labelValue)){
                                videoItem.setStars(videoItem.getStars() + 1);
                            }else if ("sub".equals(labelValue)){
                                videoItem.setStars(videoItem.getStars() - 1);
                            }
                            videoItemService.updateById(videoItem);
                        }
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    } catch (Exception e){
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }
                }
            });
            consumer.start();
        } catch (Exception e) {
            throw new JavaSmException(CodeAndMsg.MQ_CONSUMER_ERROR);
        }
        return consumer;
    }

    @Bean
    public DefaultMQPushConsumer defaultMQPushConsumer2(VideoItemService videoItemService, VideoService videoService){
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("videoCollectRecord");
        consumer.setNamesrvAddr(nameSrvAddr);
        try {
            consumer.subscribe("videoCollectRecord", "*");
            consumer.registerMessageListener(new MessageListenerConcurrently() {
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                    for (MessageExt messageExt : list) {
                        String labelValue = new String(messageExt.getUserProperty("myLabel"));
                        String str = new String(messageExt.getBody());
                        Long videoItemId = Long.parseLong(str);
                        VideoItem videoItem = videoItemService.getById(videoItemId);
                        if (videoItem == null) {
                            throw new JavaSmException(CodeAndMsg.VIDEO_ITEM_NOT_EXIST);
                        }
                        Video video = videoService.getById(videoItem.getVideoId());
                        if ("add".equals(labelValue)){
                            video.setFollowCount(video.getFollowCount() + 1);
                        }else if ("sub".equals(labelValue)){
                            video.setFollowCount(video.getFollowCount() - 1);
                        }
                        videoService.updateById(video);
                    }
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
            consumer.start();
        } catch (Exception e) {
            throw new JavaSmException(CodeAndMsg.MQ_CONSUMER_ERROR);
        }
        return consumer;
    }
}
